package com.wormpex.data.root.listener;

import com.google.common.base.Stopwatch;
import com.wormpex.api.json.JsonUtil;
import com.wormpex.data.root.dao.FailedTransmitMessageDao;
import com.wormpex.data.root.service.WmqTransmitKafkaService;
import com.wormpex.inf.wmq.api.Message;
import com.wormpex.inf.wmq.api.MessageListener;
import com.wormpex.monitor.WMonitor;
import com.wormpex.monitor.WMonitorType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.Resource;
import java.util.concurrent.TimeUnit;

/**
 * wmq订单监听器
 *
 * @author yetong.gao
 * @since 21 JUL 2017
 */
//@Component("wmqListener")
public class WmqListener implements MessageListener {

    private final static Logger LOG = LoggerFactory.getLogger(WmqListener.class);



    private WmqTransmitKafkaService wmqTransmitKafkaServiceImpl;

    @Resource
    private FailedTransmitMessageDao failedTransmitMessageDao;



    @Override
    public void onMessage(Message message) {
//        try {
            LOG.info("收到线上消息！ msg={}", JsonUtil.toJson(message));
            //记录消息耗时
            Stopwatch stopwatch = Stopwatch.createStarted();
            LOG.info("transmit_wmq_message_by_kafka_process topic ", message.getMessageId(),
                    stopwatch.elapsed(TimeUnit.MILLISECONDS));
            WMonitor.recordTransaction(WMonitorType.SERVICE, "update_realtime_sales_order_time", stopwatch.elapsed(TimeUnit.MILLISECONDS));
            wmqTransmitKafkaServiceImpl.transmit(message);

            //错误处理
//        } catch (Exception e){
//            WMonitor.recordEvent(WMonitorType.SERVICE, "update_realtime_data_failed");
//            LOG.error("update_realtime_data_failed msg:{} failed,message={}",
//                    message.getMessageId(),
//                    JsonUtil.of(message),
//                    e);
//            FailedTransmitMessage failedTransmitMessage = new FailedTransmitMessage();
//            failedTransmitMessage.setFailedType("");
//            failedTransmitMessage.setTopicName(message.getTopic());
//            failedTransmitMessage.setMessage(JsonUtil.of(message.getAttrs()));
//            failedTransmitMessageDao.insert(failedTransmitMessage);
//        }
    }
}
